package datastore.api.impl.hbase;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.io.BatchUpdate;
import org.apache.hadoop.hbase.io.Cell;

import datastore.db.Key;
import datastore.db.Model;
import datastore.api.impl.serializer.JsonSerializer;
import datastore.api.impl.serializer.Serializer;

public class EntityStore extends BaseStore<Model, Object> {
    static final String TABLE_NAME = "entities";
    private static final String COL_FAM = "entity:";
    private static final String COL_QUALIFIER = "value";
    static final String COLUMN = COL_FAM + COL_QUALIFIER;

    private Serializer serializer;
    
    public EntityStore(HBaseDatastore hbase) {
        super(hbase);
        this.serializer = new JsonSerializer();
        // this.serializer = new XStreamSerializer();
    }
    
    @Override
    public void init() throws Exception {
        List<BatchUpdate> updates = new ArrayList<BatchUpdate>();
        
        //Create table
        boolean tableExists = admin.tableExists(TABLE_NAME);
        if (!tableExists) {
            createTable(updates, TABLE_NAME, COL_FAM, COLUMN, 
                    constructKey(DEFAULT_KEY_PART, DEFAULT_KEY_PART));
        }

        table = new HTable(config, TABLE_NAME);
        /*for (String kind : kinds) {
            String rowKey = constructKey(kind, DEFAULT_KEY_PART);
            if (!tableExists || !table.exists(rowKey.getBytes())) {
                updates.add(getBatch(rowKey, COLUMN, EMPTY_VALUE));
            }
        }*/
        if (!updates.isEmpty()) {
            table.commit(updates);
        }
    }

    @Override
    public void put(Model model) throws Exception {
        Key key = null;
        if (!model.isSaved()) {
            if (model.getKeyName() == null) {
                throw new UnsupportedOperationException();
                //TODO: generate id 
                //use 0.19.1 incrementColumnValue(table,row,family:column) -> Inserts a new version of the specified column (which must already exist) with an incremented cell value.
                //  incrementFamilyColumn(table,row,family: ) - http://issues.apache.org/jira/browse/HBASE-803
            } else {
                key = Key.fromPath(model.getParentKey(), model.getClass(), model.getKeyName());
            }
            model.setKey(key);
        } else {
            key = model.getKey();
        }
        String rowKey = toEntityRowKey(key);
        byte[] bytes = serializer.toBytes(model);
        putRow(rowKey, TABLE_NAME, COLUMN, bytes);
    }

    @Override
    public void delete(Model model) throws Exception {
        String entityRowKey = toEntityRowKey(model.getKey());
        deleteRow(entityRowKey, TABLE_NAME);
    }
    
    @Override
    public Model get(Object keyObj) throws Exception {
        String rowKey;
        Key key;
        
        if (keyObj instanceof Key) {
            key = (Key) keyObj;
            rowKey = toEntityRowKey(key);
        } else if (keyObj instanceof String) {
            rowKey = (String) keyObj;
            key = toKey(rowKey);
        } else {
            throw new IllegalArgumentException("Type must be Key or String, recieved: "+keyObj.getClass());
        }
        
        Cell cell = table.get(rowKey, COLUMN);
        if (cell == null) { //no entity specified by key found
            return null;
        }
        Model model = (Model) serializer.fromBytes(cell.getValue());
        model.setKey(key);
        return model;
    }

}
